3. grpc hello world server 解析 - 《gRPC 源码解读》

这节内容我们先来研究一下 server 端连接建立过程。

先放上 server 端的 main 函数。

我们发现其实 server 端连接的建立主要包括三步：

（1）创建 server

（2）server 的注册

（3）调用 Serve 监听端口并处理请求

ok，弄清楚主流程之后下面我们进入每个步骤里面去看一下代码实现。

1、创建 server

server 的创建比较简单，其实就下面一个方法：


    opts := defaultServerOptions
    for _, o := range opt {
        o.apply(&opts)
    }
    s := &Server{
        lis:    make(map[net.Listener]bool),
        opts:   opts,
        conns:  make(map[transport.ServerTransport]bool),
        m:      make(map[string]*service),
        quit:   grpcsync.NewEvent(),
        done:   grpcsync.NewEvent(),
        czData: new(channelzData),
    }
    s.cv = sync.NewCond(&s.mu)
    if EnableTracing {
        _, file, line, _ := runtime.Caller(1)
        s.events = trace.NewEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
    }
    if channelz.IsOn() {
        s.channelzID = channelz.RegisterServer(&channelzServer{s}, "")
    }
    return s
}

// Server is a gRPC server to serve RPC requests.
type Server struct {
    // serverOptions 就是描述协议的各种参数选项，包括发送和接收的消息大小、buffer大小等等各种，跟 http Headers 类似，我们这里就暂时先不管
    opts serverOptions
    // 一个互斥锁
    mu     sync.Mutex // guards following
    // listener map
    lis    map[net.Listener]bool
    // connections map
    conns  map[transport.ServerTransport]bool
    // server 是否在处理请求的一个状态位
    serve  bool
    drain  bool
    cv     *sync.Cond          // signaled when connections close for GracefulStop
    // service map
    m      map[string]*service // service name -> service info
    events trace.EventLog
    quit               *grpcsync.Event
    done               *grpcsync.Event
    channelzRemoveOnce sync.Once
    serveWG            sync.WaitGroup // counts active Serve goroutines for GracefulStop
    channelzID int64 // channelz unique identification number
    czData     *channelzData
}

虽然 server 结构体里面各种乱起八糟的字段，但是我们可以先不管哈哈哈，比较重要的无非就是三个 map 表分别用来存放多个 listener 、connection 和 service。其他字段都是为了实现协议描述或者并发控制的功能。我们重点关注下

m      map[string]*service

这个结构，service 中主要包含了 MethodDesc 和 StreamDesc 这两个 map

type service struct {
    server interface{} // the server for service methods
    md     map[string]*MethodDesc
    sd     map[string]*StreamDesc
    mdata  interface{}
}

2、server 注册

server 的注册调用了 RegisterGreeterServer 方法，这个方法是 pb.go 文件里面的，如下：

这个方法调用了 server 的 RegisterService 方法，然后传入了一个 ServiceDesc 的数据结构，如下：

var _Greeter_serviceDesc = grpc.ServiceDesc{
    ServiceName: "helloworld.Greeter",
    HandlerType: (*GreeterServer)(nil),
    Methods: []grpc.MethodDesc{
        {
            MethodName: "SayHello",
            Handler:    _Greeter_SayHello_Handler,
        },
        {
            MethodName: "SayHelloAgain",
            Handler:    _Greeter_SayHelloAgain_Handler,
        },
    },
    Metadata: "helloworld.proto",
}

我们来看看 RegisterService 这个方法，可以看到主要是调用了 register 方法，register 方法则按照方法名为 key，将方法注入到 server 的 service map 中。看到这里我们其实可以预测一下，server 不同 rpc 请求的处理，也是根据 service 中不同的 serviceName 去 service map 中取出不同的 handler 进行处理


    ht := reflect.TypeOf(sd.HandlerType).Elem()
    st := reflect.TypeOf(ss)
    if !st.Implements(ht) {
        grpclog.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)
    }
    s.register(sd, ss)
}
func (s *Server) register(sd *ServiceDesc, ss interface{}) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.printf("RegisterService(%q)", sd.ServiceName)
    if s.serve {
        grpclog.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)
    }
    if _, ok := s.m[sd.ServiceName]; ok {
        grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
    }
    srv := &service{
        server: ss,
        md:     make(map[string]*MethodDesc),
        sd:     make(map[string]*StreamDesc),
        mdata:  sd.Metadata,
    }
    for i := range sd.Methods {
        d := &sd.Methods[i]
        srv.md[d.MethodName] = d
    }
    for i := range sd.Streams {
        d := &sd.Streams[i]
        srv.sd[d.StreamName] = d
    }
    s.m[sd.ServiceName] = srv
}

3、Serve 过程

回想所有 C/S 模式下，client 和 server 的通信基本是类似的。大致过程无非是 server 通过死循环的方式在某一个端口实现监听，然后 client 对这个端口发起连接请求，握手成功后建立连接，然后 server 处理 client 发送过来的请求数据，根据请求类型和请求参数，调用不同的 handler 进行处理，回写响应数据。

所以，对 server 端来说，主要是了解其如何实现监听，如何为请求分配不同的 handler 和回写响应数据。

for {
        rawConn, err := lis.Accept()
        ......
        s.serveWG.Add(1)
        go func() {
            s.handleRawConn(rawConn)
            s.serveWG.Done()
        }()
    }

ok，我们已经看到了监听过程，server 的监听果然是通过一个死循环调用了 lis.Accept() 进行端口监听。

继续往下看，我们发现新起协程调用了 handleRawConn 这个方法，为了节约篇幅，我们直接看重点代码，如下：

func (s *Server) handleRawConn(rawConn net.Conn) {
    ...
    conn, authInfo, err := s.useTransportAuthenticator(rawConn)
    ...
    // Finish handshaking (HTTP2)
    st := s.newHTTP2Transport(conn, authInfo)
    if st == nil {
        return
    }
    ...
    go func() {
        s.serveStreams(st)
        s.removeConn(st)
    }()
}

可以看到 handleRawConn 里面实现了 http 的 handshake，还记得之前我们说过，grpc 是基于 http2 实现的吗？这里是不是实锤了……. 发现又通过一个新的协程调用了 serveStreams 这个方法，这个方法干了啥呢？

其实它主要调用了 handleStream ，继续跟进 handleStream 方法，我们发现了重要线索，如下（省略了部分无关代码）

func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream, trInfo *traceInfo) {
    sm := stream.Method()
    ...
    service := sm[:pos]
    method := sm[pos+1:]
    srv, knownService := s.m[service]
        if md, ok := srv.md[method]; ok {
            s.processUnaryRPC(t, stream, srv, md, trInfo)
            return
        }
        if sd, ok := srv.sd[method]; ok {
            s.processStreamingRPC(t, stream, srv, sd, trInfo)
            return
        }
    }
    ...
}

重要线索就是这一行

    srv, knownService := s.m[service]

还记得我们之前的预测吗？根据 serviceName 去 server 中的 service map，也就是 m 这个字段，里面去取出 handler 进行处理。我们 hello world 这个 demo 的请求不涉及到 stream ，所以直接取出 handler ，然后传给 processUnaryRPC 这个方法进行处理。

if md, ok := srv.md[method]; ok {
    s.processUnaryRPC(t, stream, srv, md, trInfo)
    return
}

再来看看 processUnaryRpc 这个方法

func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
    ...
    sh := s.opts.statsHandler
    if sh != nil {
        beginTime := time.Now()
        begin := &stats.Begin{
            BeginTime: beginTime,
        }
        sh.HandleRPC(stream.Context(), begin)
        defer func() {
            end := &stats.End{
                BeginTime: beginTime,
                EndTime:   time.Now(),
            }
            if err != nil && err != io.EOF {
                end.Error = toRPCErr(err)
            }
            sh.HandleRPC(stream.Context(), end)
        }()
    }
    ...
    if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {
        if err == io.EOF {
            // The entire stream is done (for unary RPC only).
            return err
        }
        if s, ok := status.FromError(err); ok {
            if e := t.WriteStatus(stream, s); e != nil {
                grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
            }
        } else {
            switch st := err.(type) {
            case transport.ConnectionError:
                // Nothing to do here.
            default:
                panic(fmt.Sprintf("grpc: Unexpected error (%T) from sendResponse: %v", st, st))
            }
        }
        if binlog != nil {
            h, _ := stream.Header()
            binlog.Log(&binarylog.ServerHeader{
                Header: h,
            })
            binlog.Log(&binarylog.ServerTrailer{
                Trailer: stream.Trailer(),
                Err:     appErr,
            })
        }
        return err
    }
    ...

我们终于看到了 handler 对 rpc 的处理：

至此，server 端我们的目标实现，追踪到了整个请求和监听、handler 处理和 response 回写的过程。